package org.llc.flink.logs

import java.util.Properties

import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.streaming.util.serialization.SimpleStringSchema
import org.llc.flink.logs.utils.ParamsConf

object KafkaConsumer {

  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    //配置
    val properties = new Properties()
    properties.setProperty("group.id", ParamsConf.kafkaGroupId)
    properties.setProperty("bootstrap.servers", ParamsConf.kafkaHosts)
    properties.setProperty("key.deserializer",
      "org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty("value.deserializer",
      "org.apache.kafka.common.serialization.StringDeserializer")
    //获取数据
    val ds: DataStream[String] = env.addSource(new FlinkKafkaConsumer[String](ParamsConf.kafkaTopic, new SimpleStringSchema(), properties))
    //输出
    ds.print()
    //执行
    env.execute("kafkaSource test")

  }

}
